package cn.test.spark_streaming

import kafka.serializer.StringDecoder
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @Author: Cyy
  * @Description: Spark Streaming对接Kafka的方式-----通过kafka broker
  * @Date:Created in 23:22 2019/3/24
  */
object kd_spark_streaming{
  def main(args: Array[String]): Unit = {

    if(args.length != 2) {
      System.err.println("Usage: KafkaDirectWordCount <brokers> <topics>")
      System.exit(1)
    }
    val Array(brokers, topics) = args

    val sparkConf: SparkConf = new SparkConf()
      .setAppName("SparkStreamingKafka_Direct").setMaster("local[2]")

    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")

    val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
//    ssc.checkpoint("SparkStreaming/kafka_direct")

    val kafkaParams = Map("metadata.broker.list"->brokers)

    val topics1: Set[String] = Set(topics)
//    val topicsSet = topics.split(",").toSet

    val dstream: InputDStream[(String,String)] = KafkaUtils
      .createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics1)

    dstream.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

    ssc.start()

    ssc.awaitTermination()
  }

}
